package com.yunhe.abnormal.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.AbnormalConsts;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.abnormal.domain.data.AlarmEvent;
import com.yunhe.common.model.msg.AlarmEventMsgDTO;
import com.yunhe.common.model.PageParam;
import com.yunhe.common.model.base.AbnormalStatusDTO;
import com.yunhe.common.model.data.AlarmEventDTO;
import com.yunhe.abnormal.service.base.AbnormalStatusService;
import com.yunhe.abnormal.service.data.AlarmEventService;
import com.yunhe.common.util.CollectionUtil;
import com.yunhe.common.util.JsonUtil;
import com.yunhe.common.util.StringUtil;
import com.yunhe.common.websocket.WebsocketRealtimeHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 实时异常WebSocket处理类
 *
 * @author liuronglei
 */
public class AlarmOverviewHandler extends WebsocketRealtimeHandler {
    private static final Logger logger = LoggerFactory.getLogger(AlarmOverviewHandler.class);
    private ObjectMapper mapper = JsonUtil.getObjectMapper();
    @Autowired
    private AlarmEventService alarmEventService;
    @Autowired
    private AbnormalStatusService abnormalStatusService;
    private PageParam pageParam;

    @PostConstruct
    void init() {
        // 初始化分页参数，为了避免数据量过大，只取前100条记录
        pageParam = new PageParam();
        pageParam.setPage(1);
        pageParam.setSize(100);
    }

    @Override
    public Long refreshInterval() {
        return 5000L;
    }

    @Override
    public void beforeSend(WebSocketSession session, Object content) {

    }

    @Override
    public String getPointField() {
        return "stationIds";
    }

    @Override
    public void afterEstablished(WebSocketSession session) {

    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {

    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points) {
        Map<String, List<AlarmEvent>> result = getAlarmEvents(points);
        sendMessage(result, session, points);
    }

    /**
     * 获得需要展示的告警状态ID
     */
    private List<Long> getAbnormalStatusIds() {
        List<Long> abnormalStatusIds = new ArrayList<>();
        AbnormalStatusDTO statusReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.READY.getName());
        if (statusReady != null) {
            abnormalStatusIds.add(statusReady.getId());
        }
        AbnormalStatusDTO sentReady = abnormalStatusService.findByName(AbnormalConsts.ABNORMAL_STATUS.SENT.getName());
        if (sentReady != null) {
            abnormalStatusIds.add(sentReady.getId());
        }
        return abnormalStatusIds;
    }

    /**
     * 获得实时告警列表
     * @param points key列表
     */
    private Map<String, List<AlarmEvent>> getAlarmEvents(List<String> points) {
        Map<String, List<AlarmEvent>> result = new HashMap<>();
        if (CollectionUtil.isNotEmpty(points)) {
            List<Long> abnormalStatusIds = getAbnormalStatusIds();
            List<Long> stations = new ArrayList<>(points.size());
            points.forEach(point -> stations.add(Long.parseLong(point)));
            Iterable<AlarmEvent> alarmEvents = alarmEventService.findByCondition(null, stations, null, null,
                    null, abnormalStatusIds, null, null, null, pageParam);
            if (alarmEvents != null) {
                alarmEvents.forEach(alarmEvent -> {
                    String point = String.valueOf(alarmEvent.getStationId());
                    List<AlarmEvent> alarmEventList = result.computeIfAbsent(point, list -> new ArrayList<>());
                    alarmEventList.add(alarmEvent);
                });
            }
        }
        // 如果电站没有告警，也要加一个空列表
        points.forEach(point -> result.computeIfAbsent(point, list -> new ArrayList<>()));
        return result;
    }

    /**
     * 监听告警内容刷新消息
     * @param record 消息内容：主要是电站ID
     */
    @KafkaListener(topics = KafkaConsts.Topic.AlarmEventRefresh, groupId = KafkaConsts.Group.RandomGroupId)
    public void overviewListener(ConsumerRecord<?, String> record) {
        try {
            String value = record.value();
            if (!StringUtil.isEmpty(value)) {
                AlarmEventMsgDTO msg = mapper.readValue(value, AlarmEventMsgDTO.class);
                List<AlarmEventDTO> alarmEvents = msg.getAlarmEvents();
                List<String> stations = new ArrayList<>();
                for (AlarmEventDTO alarmEvent : alarmEvents) {
                    String stationId = String.valueOf(alarmEvent.getStationId());
                    if (!stations.contains(stationId)) {
                        stations.add(stationId);
                    }
                }
                Map<String, List<AlarmEvent>> result = getAlarmEvents(stations);
                sendMessage(result, null);
            }
        } catch (Exception e) {
            logger.error("电站实时告警推送报错，{}，{}", record.value(), e);
        }
    }
}